package com.bosi.message.impl;

import com.bosi.message.Message;
import com.bosi.message.MessageService;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

@Slf4j
public class SimpleMessageService implements MessageService {
    protected final ConcurrentHashMap<String,Handler> map = new ConcurrentHashMap<>();

    public SimpleMessageService(){
        log.info("Simple Message Service start.");
    }

    @Override
    public void publish(Message message) {
        map.values()
                .stream()
                .filter(e -> isContain(message.getHeader(), e.header))
                .map(e -> e.consumer)
                .forEach(e -> e.accept(message));
    }

    /**
     * 大map是否包含小map
     * @param bigMap
     * @param smallMap
     * @return
     */
    public boolean isContain(Map<String,Object> bigMap,Map<String,Object> smallMap){
        if(smallMap.size() == 0) return true;

        for (Map.Entry<String, Object> entry : smallMap.entrySet()) {
            if (!entry.getValue().equals(bigMap.get(entry.getKey()))) {
                return false;
            }
        }

        return true;
    }

    @Override
    public String subscribe(Map<String,Object> header, Consumer<Message> consumer) {
        String uuid = UUID.randomUUID().toString();
        map.put(uuid,new Handler(header,consumer));
        log.debug("subscribe channel size = {}",map.size());
        return uuid;
    }

    @Override
    public void unsubscribe(String id) {
        if(id != null){
            map.remove(id);
            log.debug("unsubscribe channel size = {}",map.size());
        }
    }

    public static class Handler{
        Map<String,Object> header;
        Consumer<Message> consumer;

        public Handler(Map<String, Object> header, Consumer<Message> consumer) {
            this.header = header;
            this.consumer = consumer;
        }
    }
}
